package com.hopu.user.time;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.time.TimeTop3Cate;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class TimeTop3CateAna {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("").setMaster("local");

        JavaSparkContext context = new JavaSparkContext(conf);
        //加载日志文件
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        //获取日志文件中的hour、cid、name
        JavaPairRDD<TimeTop3Cate, Integer> top3CateRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String hour = json.getJSONObject("odate").getString("hour");
            String type = json.getString("event_type");
            String cid = json.getString("category_id");
            String cname = json.getString("category_name");

            return new Tuple2<>(new TimeTop3Cate(hour, type, cid, cname), 1);
        });

        //过滤每个时间段的RDD
        for (int i = 0; i <= 23; i++) {
            transAndPut(top3CateRDD, i+"");
        }
    }

    public static void transAndPut(JavaPairRDD<TimeTop3Cate, Integer> top3CateRDD, String hour) {

        JavaPairRDD<TimeTop3Cate, Integer> hourRDD = top3CateRDD.filter(t -> hour.equals(t._1.getHour()));
        JavaPairRDD<TimeTop3Cate, Integer> viewRDD = hourRDD.filter(t -> "view".equals(t._1.getType()));
        JavaPairRDD<TimeTop3Cate, Integer> cartRDD = hourRDD.filter(t -> "cart".equals(t._1.getType()));
        JavaPairRDD<TimeTop3Cate, Integer> parchaseRDD = hourRDD.filter(t -> "parchase".equals(t._1.getType()));

        JavaPairRDD<TimeTop3Cate, Integer> viewTotalRDD = groupByCid(viewRDD);
        JavaPairRDD<TimeTop3Cate, Integer> cartTotalRDD = groupByCid(cartRDD);
        JavaPairRDD<TimeTop3Cate, Integer> parchaseTotalRDD = groupByCid(parchaseRDD);

        JavaPairRDD<TimeTop3Cate, Integer> top3AllCount = combinRDD(viewTotalRDD, cartTotalRDD, parchaseTotalRDD);

        //排序
        top3AllCount = top3AllCount.sortByKey();

        //取出某个时间段，top3的热门类目
        List<TimeTop3Cate> top3Cates = top3AllCount.map(t -> t._1).take(3);
        top3Cates.forEach(t -> System.out.println(t));

        //把top3类目保存到数据库
        String[] columns = {"hour", "cid", "cname", "viewCount", "cartCount", "parchaseCount"};

        for (TimeTop3Cate top3Cate: top3Cates) {
            String[] values = {top3Cate.getHour(), top3Cate.getCid(), top3Cate.getCname(), top3Cate.getViewCount()+"",
                                top3Cate.getCartCount()+"", top3Cate.getParchaseCount()+""};
            try {
                HBaseUtils.putOneRowToHbase("time", top3Cate.getHour(), "top3cate", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static JavaPairRDD<TimeTop3Cate, Integer> combinRDD(JavaPairRDD<TimeTop3Cate, Integer> viewTotalRDD,
                                 JavaPairRDD<TimeTop3Cate, Integer> cartTotalRDD,
                                 JavaPairRDD<TimeTop3Cate, Integer> parchaseTotalRDD) {
        //合并3个类型的RDD，左外连接进行合并
        JavaPairRDD<TimeTop3Cate, Tuple2<Integer, Optional<Integer>>> result = viewTotalRDD.leftOuterJoin(cartTotalRDD);
        JavaPairRDD<TimeTop3Cate, Integer> top3ViewAndCartCount = result.mapToPair(t -> {
            TimeTop3Cate top3Cate = t._1;//取出对象

            top3Cate.setCartCount(t._2._2.get());

            return new Tuple2<>(top3Cate, top3Cate.getViewCount()+top3Cate.getCartCount());
        });
//        top3ViewAndCartCount.foreach(t -> System.out.println(t));

        JavaPairRDD<TimeTop3Cate, Tuple2<Integer, Optional<Integer>>> timeTop3CateTuple2JavaPairRDD = top3ViewAndCartCount.leftOuterJoin(parchaseTotalRDD);
        JavaPairRDD<TimeTop3Cate, Integer> top3AllCount = timeTop3CateTuple2JavaPairRDD.mapToPair(t -> {
            TimeTop3Cate top3Cate = t._1;
            Tuple2<Integer, Optional<Integer>> tuple = t._2;//取出值

            int viewAndCartCount = tuple._1;

            Optional<Integer> parchaseOption = tuple._2;
            Integer parchaseCount = parchaseOption.get();

            top3Cate.setParchaseCount(parchaseCount);

            return new Tuple2<>(top3Cate, viewAndCartCount + parchaseCount);
        });
//        top3AllCount.foreach(t -> System.out.println(t));
        return top3AllCount;
    }

    public static JavaPairRDD<TimeTop3Cate, Integer> groupByCid(JavaPairRDD<TimeTop3Cate, Integer> hourAndTypeRDD) {
        JavaPairRDD<TimeTop3Cate, Iterable<Integer>> groupByCid = hourAndTypeRDD.groupByKey();

        JavaPairRDD<TimeTop3Cate, Integer> typeByCidGroup = groupByCid.mapToPair(t -> {
            int count = 0;
            TimeTop3Cate top3Cate = t._1;
            for (int v : t._2) {
                count += v;
            }
            if (top3Cate.getType().equals("view")) {
                top3Cate.setViewCount(count);
            } else if (top3Cate.getType().equals("cart")) {
                top3Cate.setCartCount(count);
            } else {
                top3Cate.setParchaseCount(count);
            }
            return new Tuple2<>(top3Cate, count);
        });

//        typeByCidGroup.foreach(t -> System.out.println(t));

        return typeByCidGroup;


    }

}
